package com.xzx.flink.streamapi.mutilstream;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/5/5 18:30
 */
public class ConnectedStream_BillCheck {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Tuple3<String, String, Long>> appStream = env.fromElements(
                        Tuple3.of("order-1", "app", 1000L),
                        Tuple3.of("order-2", "app", 2000L))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Long>>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
                            @Override
                            public long extractTimestamp(Tuple3<String, String, Long> element, long recordTimestamp) {
                                return element.f2;
                            }
                        }));


        SingleOutputStreamOperator<Tuple4<String, String, String, Long>> thirdStream = env.fromElements(
                        Tuple4.of("order-1", "third-party", "success", 3000L),
                        Tuple4.of("order-3", "third-party", "success", 4000L))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple4<String, String, String, Long>>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<Tuple4<String, String, String, Long>>() {
                            @Override
                            public long extractTimestamp(Tuple4<String, String, String, Long> element, long recordTimestamp) {
                                return element.f3;
                            }
                        }));


        appStream.connect(thirdStream)
                .keyBy(d -> d.f0, d -> d.f0)
                .process(new CoProcessFunction<Tuple3<String, String, Long>, Tuple4<String, String, String, Long>, String>() {
                    private ValueState<Tuple3<String, String, Long>> appState;
                    private ValueState<Tuple4<String, String, String, Long>> thirdState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        appState = getRuntimeContext().getState(new ValueStateDescriptor<>("app", Types.TUPLE(Types.STRING, Types.STRING, Types.LONG)));
                        thirdState = getRuntimeContext().getState(new ValueStateDescriptor<>("third", Types.TUPLE(Types.STRING, Types.STRING, Types.STRING, Types.LONG)));
                    }

                    // app的消息过来，判断第三方的事件是否已经到来过
                    @Override
                    public void processElement1(Tuple3<String, String, Long> value, CoProcessFunction<Tuple3<String, String, Long>, Tuple4<String, String, String, Long>, String>.Context ctx, Collector<String> out) throws Exception {
                        if (thirdState.value() != null) {
                            //第三方的事件已经存在
                            out.collect("对账成功， app value:" + value + ";third value:" + thirdState.value());
                            thirdState.clear();
                        } else {// 第三方的事件还没来，先记录app的事件
                            appState.update(value);
                            // 注册一个5秒的定时器来检测第三方的事件是都到来
                            ctx.timerService().registerEventTimeTimer(value.f2 + 5000L);
                        }
                    }

                    @Override
                    public void processElement2(Tuple4<String, String, String, Long> value, CoProcessFunction<Tuple3<String, String, Long>, Tuple4<String, String, String, Long>, String>.Context ctx, Collector<String> out) throws Exception {
                        if (appState.value() != null) {
                            out.collect("对账成功， app value:" + appState.value() + ";third value:" + value);
                            appState.clear();
                        } else {
                            thirdState.update(value);
                            ctx.timerService().registerEventTimeTimer(value.f3 + 5000L);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, CoProcessFunction<Tuple3<String, String, Long>, Tuple4<String, String, String, Long>, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        if (appState.value() != null) {// 5秒之后，app的状态还在，说明队长失败了，第三方的事件没有在5秒内收到
                            out.collect("对账失败：" + appState.value() + " 第三方支付平台信息未到");
                        }
                        if (thirdState.value() != null) {
                            out.collect("对账失败：" + thirdState.value() + " app信息未到");
                        }
                        appState.clear();
                        thirdState.clear();
                    }
                })
                .print();

        env.execute(ConnectedStream_BillCheck.class.getSimpleName());
    }
}
